package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/4/14
 * @Desc: 学大数据，上多易教育
 *   对kafka中的用户行为日志记录，统计每5分钟每个页面的访问次数
 *
 *
{"uid":1,"event_id":"page_load","properties":{"url":"/a","ref":"/x"},"action_time":1704719574000}
{"uid":1,"event_id":"add_cart","properties":{"url":"/a","item_id":"p01"},"action_time":1704719575000}
{"uid":1,"event_id":"page_load","properties":{"url":"/b","ref":"/a"},"action_time":1704719576000}
{"uid":2,"event_id":"page_load","properties":{"url":"/a","ref":"/x"},"action_time":1704719577000}
{"uid":2,"event_id":"video_play","properties":{"url":"/a","video_id":"v01"},"action_time":1704719578000}
{"uid":2,"event_id":"page_load","properties":{"url":"/b","ref":"/x"},"action_time":1704719579000}

{"uid":1,"event_id":"page_load","properties":{"url":"/c","ref":"/x"},"action_time":1704720581000}
 *
 *
 **/
public class Demo4_TumbleWindowAgg {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射物理数据源
        tenv.executeSql(
                        " create table user_events(                            "+
                        "     uid bigint,                                      "+
                        "     event_id string,                                 "+
                        "     properties map<string,string>,                   "+
                        "     action_time bigint,                              "+
                        "     url as properties['url'],                        "+
                        "     pt as proctime(),                                "+
                        "     rt as to_timestamp_ltz(action_time,3),           "+
                        "     watermark for rt as rt,                          "+
                        "     partition_id bigint metadata from 'partition'    "+
                        " ) with (                                             "+
                        "   'connector' = 'kafka',                             "+
                        "   'topic' = 'tpc-a',                                 "+
                        "   'properties.bootstrap.servers' = 'doitedu:9092',   "+
                        "   'properties.group.id' = 'doit47-g2',               "+
                        "   'scan.startup.mode' = 'latest-offset',             "+
                        "   'value.format' = 'json',                           "+
                        "   'value.fields-include' = 'EXCEPT_KEY'              "+
                        " )                                                    "
        );

        // 统计每5分钟每个页面的访问次数
        tenv.executeSql(
                        " SELECT                                                              "+
                        " window_start,                                                       "+
                        " window_end,                                                         "+
                        " url,                                                                "+
                        // 非法使用null
                        // count(if(event_id='page_load',1,null)) as pv
                        // sum(if(event_id='page_load',1,0)) as pv
                        // 推荐写法，可以起到谓词下推的优化效果
                        " count(event_id) filter(where event_id = 'page_load') as pv          "+
                        "                                                                     "+
                        " from table(                                                         "+
                        "     tumble(table user_events, descriptor(rt), interval '5' minute)  "+
                        " )                                                                   "+
                        " group by                                                            "+
                        " window_start,                                                       "+
                        " window_end,                                                         "+
                        " url                                                                 "
        ).print();



    }
}
